package com.learning.thread.design.pc;

import com.google.common.base.Strings;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * ClassName: Consumer
 * Description:  消费者
 *
 * @author sm12652
 * @version V1.0
 * @since JDK 1.7
 */
public class Consumer implements Runnable {
    private final List<String> valueList = new ArrayList<>();
    private final ConcurrentLinkedQueue<String> queue;
    private static final int BATCH_FLUSH_NUM = 20;

    public Consumer(ConcurrentLinkedQueue<String> queue) {
        this.queue = queue;
    }

    private volatile boolean runFlag = true;

    public void shutDownThread() {
        runFlag = false;
    }

    @Override
    public void run() {
        long start = System.currentTimeMillis();
        while (runFlag) {
            try {
                String msg = queue.poll();
//               queue.peek() 返回但不移除队列头部的元素
                if (!Strings.isNullOrEmpty(msg)) {
                    valueList.add(msg);
                } else {
                    Thread.sleep(2000L);
                }
                long end = System.currentTimeMillis();
                if ((valueList.size() == BATCH_FLUSH_NUM) || ((end - start) > 3 * 1000 && !valueList.isEmpty())) {
                    System.out.println("拿数据干活~~~");
                    valueList.clear();
                    start = end;
                }
            } catch (Exception e) {
            }
        }
    }
}
